package zjy.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by Jiyang.Zheng on 2022/3/3 14:29.
 */
public class WordCount {
    public static final String INPUT_PATH = "hdfs://192.168.253.200:9000/test/words.txt";
    public static final String OUTPUT_PATH = "hdfs://192.168.253.200:9000/test/output/wordcount";

    public static class TokenizerMapper
            extends Mapper<Object, Text, Text, IntWritable>{//定义MAP的输入输出的Key-Value类型，第一对key-value的key不一定是字符串，所以申明为Object类型

        private final static IntWritable one = new IntWritable(1);//key-value中经常用到1，申明为常量
        private Text word = new Text();//相当于java中的String

        //    map方法每次只处理一行【应该是的，我在另一个程序中打印输出是每次一行，还待深究】
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {//这里三个变量，前两个对应上面的，context封装了输出的key-value
            StringTokenizer itr = new StringTokenizer(value.toString());//字符串分词器
//      String[] strs = value.toString().split(" "); //等于上面这行，就是分词
            while (itr.hasMoreTokens()) {// 取出子串，并赋1，写入context。不用管context存在什么地方
                word.set(itr.nextToken());
                context.write(word, one);
                //有个概念要知道，这里他给每个词都赋值value=1，实际上value=2也是可以的（
                // 比如hello hello，可以一次计算value=2），看你怎么处理了(value=1省去了一步相同词再汇总过程)
            }
        }
    }

    public static class IntSumReducer
            extends Reducer<Text,IntWritable,Text,IntWritable> {//如上，不过输入、出都是字符串-Int型
        private IntWritable result = new IntWritable();//对每个key，求value的和为result，比如<hello,3>

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();// get是将IntWritable转为Int类型，set相反，见下面
            }
            result.set(sum);// sum是Int类型，hadoop无法将之序列化，所以转为IntWritable类型
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
//        if (otherArgs.length < 2) {
//            System.err.println("Usage: wordcount <in> [<in>...] <out>");
//            System.exit(2);
//        }
        // 作业调度
        Job job = new Job(conf, "word count");
        job.setJarByClass(WordCount.class);// 指定作业调度的类
        job.setMapperClass(TokenizerMapper.class);//指定Mapper类
        job.setCombinerClass(IntSumReducer.class);//指定Combiner类
        job.setReducerClass(IntSumReducer.class);//指定Reducer类
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        for (int i = 0; i < otherArgs.length - 1; ++i) {    //传入的文件参数可为多个【最后一个是输出文件】
            FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
        }
        FileOutputFormat.setOutputPath(job,
                new Path(OUTPUT_PATH));
        System.exit(job.waitForCompletion(true) ? 0 : 1);//等待执行，0是正常退出，1是错误退出
    }
}
